/*
============================================================================
 Name        : dd_task_worker.c
 Author      : Jian
 Version     :
 Copyright   : GPL
 Description : dd_task_worker.c - source file
============================================================================
*/

/* INCLUDE FILES */
#include "dd_task_worker.h"
#include "dd_vector.h"
#include "dd_log.h"

#include "sd_platform.h"

#define __CANCEL_TASK  ((void*)-1)

static dd_task* _get_task(dd_task_worker* worker, int threadIndex, dd_task* preTask) ;

typedef struct {
    dd_task_worker* worker ;
    int             threadIndex ;
} dd_worker_thread_data;

static int _work_thread(void* ptr)
{
    dd_task_worker* worker = NULL ;
    int             threadIndex = 0 ;
    dd_task*        task = NULL ;

    __DD_ASSERT_DEFENSE(ptr, return -1) ;

    worker = ((dd_worker_thread_data*)ptr)->worker ;
    threadIndex = ((dd_worker_thread_data*)ptr)->threadIndex ;

    while(1) {
        
        task = _get_task(worker, threadIndex, task) ;
        __DD_ASSERT_DEFENSE(task, return -1) ;
        
        if (task == __CANCEL_TASK) {
            sd_post_sem(worker->cancelSem) ;
            return -1 ;
        }
        
        task->work(task) ;
    }

    dd_free(ptr) ;
    return 0 ;
}

static void _task_destroy(void* ptr) 
{
    dd_task* task = (dd_task*)ptr;
    __DD_ASSERT_DEFENSE(task, return) ;
    
    if (task == __CANCEL_TASK) 
        return ;

    task->destroy(task) ;
}

static int _cancel_thread(void* ptr) 
{
    dd_task_worker* worker = (dd_task_worker*)ptr ;
    int             i = 0 ;

    __DD_ASSERT_DEFENSE(worker, return -1) ;
    
    sd_wait_rlock(worker->listLock) ;

    dd_vector_reset(worker->taskList, _task_destroy) ;

    for (i = 0; i < dd_vector_count(worker->threadList); i++) {
        dd_vector_insert(worker->taskList, 0, __CANCEL_TASK) ;
        sd_post_sem(worker->waitSem) ;
    }
    
    sd_post_rlock(worker->listLock) ;

    for (i = 0; i < dd_vector_count(worker->threadList); i++) {
        if (sd_wait_sem(worker->cancelSem) < 0) 
            return -1 ;
    }

    sd_sleep(2) ;
    
    sd_destroy_rlock(worker->listLock) ;
    sd_destroy_sem(worker->cancelSem) ;
    sd_destroy_sem(worker->waitSem) ;
    
    dd_vector_destroy(worker->taskList, NULL) ;
    dd_vector_destroy(worker->threadList, sd_thread_destroy) ;
    
    dd_free(worker->curTasks) ;
    
    dd_free(worker) ;
    return 0 ;
}

static dd_task* _get_task(dd_task_worker* worker, int threadIndex, dd_task* preTask) 
{
    dd_task* task = NULL ;
    int      index = 0 ;

    sd_wait_sem(worker->waitSem) ;

    sd_wait_rlock(worker->listLock) ;
    
    __DD_ASSERT_DEFENSE(preTask == worker->curTasks[threadIndex], goto _end) ;

    if (preTask) {
        worker->curTasks[threadIndex] = NULL;
        preTask->destroy(preTask) ;

    }

    if (worker->isEnablePriority) {
        int i = 0 ;
        int highPriority = -1 ;
        int tmpPriority = -1 ;
        for (; i < dd_vector_count(worker->taskList); i++) {
            dd_task* tmpTask = (dd_task*)dd_vector_get(worker->taskList, i) ;

            if (tmpTask == __CANCEL_TASK) {
                task = (dd_task*)dd_vector_delete_by_index(worker->taskList, index) ;
                goto _end ;
            }

            tmpPriority = tmpTask->priority(tmpTask) ;
            if (tmpPriority > highPriority) {
                highPriority = tmpPriority ;
                index = i ;
            }
        }
    }

    task = (dd_task*)dd_vector_delete_by_index(worker->taskList, index) ;
    
    worker->curTasks[threadIndex] = task ;
    
    _end:

    sd_post_rlock(worker->listLock) ;

    return task ;
}

dd_task_worker* dd_worker_create(DD_UINT startTaskId, int threadCount, int isEnablePriority) 
{
    dd_task_worker* worker = NULL ;
    int             i = 0 ;

    __DD_ASSERT_DEFENSE(threadCount > 0, threadCount = 2) ;

    worker = dd_malloc(sizeof(dd_task_worker)) ;
    __DD_CHECK_MEM(worker, sizeof(dd_task_worker), return NULL) ;

    worker->taskList = dd_vector_create(8) ;
    __DD_ASSERT_DEFENSE(worker->taskList, goto _error) ;
    
    worker->curTasks = (dd_task**)dd_malloc(sizeof(dd_task*) * threadCount) ;
    __DD_CHECK_MEM(worker->curTasks, sizeof(void*) * threadCount, goto _error) ;

    worker->threadList = dd_vector_create(threadCount) ;
    __DD_ASSERT_DEFENSE(worker->threadList, goto _error) ;
    
    worker->listLock = sd_create_rlock() ;
    __DD_ASSERT_DEFENSE(worker->listLock, goto _error) ;

    worker->waitSem = sd_create_sem(0) ;
    __DD_ASSERT_DEFENSE(worker->waitSem, goto _error) ;

    worker->cancelSem = sd_create_sem(0) ;
    __DD_ASSERT_DEFENSE(worker->cancelSem, goto _error) ;

    worker->isEnablePriority = isEnablePriority ;
    worker->incTaskId        = startTaskId ;

    for (; i < threadCount; i++) {
        dd_worker_thread_data* data = NULL;
        sd_thread*             thread = NULL ;
        
        data = (dd_worker_thread_data*)dd_malloc(sizeof(dd_worker_thread_data)) ;
        __DD_ASSERT_DEFENSE(data, goto _error) ;
        data->threadIndex = i ;
        data->worker = worker ;

        thread = sd_thread_create(_work_thread, data) ; ;
        __DD_ASSERT_DEFENSE(thread, goto _error) ;

        dd_vector_append(worker->threadList, thread) ;
    }

    return worker ;

    _error:
    dd_worker_destroy(worker) ;
    return NULL ;
}

void dd_worker_destroy(dd_task_worker* worker) 
{
    __DD_ASSERT_DEFENSE(worker, return) ;

    /* TODO: the cancel thread is leak.yet */
    sd_thread_create(_cancel_thread, worker) ;
}

DD_UINT dd_worker_add(dd_task_worker* worker, dd_task* task) 
{
    int taskId = 0;

    __DD_ASSERT_DEFENSE(worker && task, return DD_INVALID_TASK_ID) ;

    sd_wait_rlock(worker->listLock) ;
    dd_vector_append(worker->taskList, task) ;
    taskId   = worker->incTaskId ;
    task->id = worker->incTaskId ;
    worker->incTaskId++ ;
    sd_post_rlock(worker->listLock) ;

    sd_post_sem(worker->waitSem) ;
    return taskId ;
}

int dd_worker_update(dd_task_worker* worker, DD_UINT taskId, void taskUpdate(dd_task*)) 
{
    int ret = -1 ;
    int i = 0 ;

    __DD_ASSERT_DEFENSE(worker && taskId != DD_INVALID_TASK_ID && taskUpdate, return -1) ;

    sd_wait_rlock(worker->listLock) ;
    
    for (; i < dd_vector_count(worker->threadList); i++) {
        if (worker->curTasks[i] && worker->curTasks[i]->id == taskId) {
            taskUpdate(worker->curTasks[i]) ;
            ret = 0 ;
            break; 
        }
    }
    
    if (ret == -1) {
        for (i = 0; i < dd_vector_count(worker->taskList); i++) {
            dd_task* task = dd_vector_get(worker->taskList, i) ;
            __DD_ASSERT_DEFENSE(task, return -1) ;
            if (task->id == taskId) {
                taskUpdate(task) ;
                ret = 0 ;
                break; 
            }
        }
    }

    sd_post_rlock(worker->listLock) ;

    return ret ;
}

/* End of File */
